package com.deepseek.project.tool;

import com.deepseek.project.constant.Constants;
import com.deepseek.project.service.ITalkInfoService;
import com.google.gson.Gson;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static java.util.concurrent.TimeUnit.MINUTES;
import static java.util.concurrent.TimeUnit.SECONDS;


/**
 * author: hulei
 * 抽象的DeepSeek工具类,为什么要抽象呢,因为笔者既想使用本地部署的deepSeek,又想使用线上官方提供的API，所以抽象了DeepSeek工具类，提取了共性
 */
@Slf4j
@Data
public abstract class AbstractChatTool {

    private static final int MAX_CONCURRENT = 20;

    private static final int BATCH_SIZE = 20;

    public static OkHttpClient client;

    private static ExecutorService httpExecutor;

    public static final Gson gson = new Gson();

    /**
     * deepSeek会话历史记录
     */
    public final List<Map<String, String>> conversationHistory = Collections.synchronizedList(new ArrayList<>());

    /**
     * 是否停止发送标志，由前端传值控制
     */
    public volatile boolean stopSending = false;

    /**
     * DeepSeek流式响应消息缓存列表，攒到一定数量时，批量保存到数据库
     */
    private final List<Map<String, String>> deepSeekMessageCache = new ArrayList<>();

    /**
     * ITalkInfoService
     */
    private final ITalkInfoService italkInfoService;

    public AbstractChatTool(ITalkInfoService italkInfoService, List<Map<String, String>> conversationHistory) {
        this.italkInfoService = italkInfoService;
        this.conversationHistory.addAll(conversationHistory);
        init();
    }

    public void init() {
        httpExecutor = Executors.newCachedThreadPool(r -> {
            Thread t = new Thread(r);
            t.setDaemon(true);
            return t;
        });

        client = new OkHttpClient.Builder()
                .dispatcher(new Dispatcher(httpExecutor))
                .connectionPool(new ConnectionPool(50, 5, MINUTES))
                .connectTimeout(30, SECONDS)
                .readTimeout(60, SECONDS)
                .writeTimeout(60, SECONDS)
                .build();
    }

    public void processMessage(String userMessage, WebSocketSession session, String talkInfoId) {
        Map<String, String> userMessageMap = Map.of("role", "user", "content", userMessage);
        synchronized (conversationHistory) {
            conversationHistory.add(userMessageMap);
        }
        if (!talkInfoId.startsWith(Constants.TEMP_TALK_INFO_ID_PREFIX)) {
            addMessageToCache(userMessageMap, talkInfoId);
        }
        requestDeepSeekMessage(session, talkInfoId);
    }


    /**
     * 构建请求头抽象方法,交给子类实现
     */
    protected abstract Request buildRequest();

    /**
     * 请求DeepSeek深度搜索消息, 交给子类实现
     */
    public abstract void requestDeepSeekMessage(WebSocketSession session, String talkInfoId);

    /**
     * 添加消息到DeepSeek流式响应消息缓存列表中
     */
    public void addMessageToCache(Map<String, String> deepSeekMessageMap, String talkInfoId) {
        deepSeekMessageCache.add(deepSeekMessageMap);
        if (deepSeekMessageCache.size() >= BATCH_SIZE) {
            saveCachedDeepSeekMessages(talkInfoId);
        }
    }

    /**
     * 保存DeepSeek流式响应消息缓存列表中的消息到数据库
     */
    public void saveCachedDeepSeekMessages(String talkInfoId) {
        if (!deepSeekMessageCache.isEmpty()) {
            try {
                italkInfoService.saveTalkInfoDeepSeekHistory(Integer.parseInt(talkInfoId), deepSeekMessageCache);
                log.info("批量消息保存到数据库成功，数量: {}", deepSeekMessageCache.size());
                deepSeekMessageCache.clear();
            } catch (Exception e) {
                log.error("批量消息保存到数据库失败，数量: {}", deepSeekMessageCache.size(), e);
            }
        }
    }

    /**
     * 发送响应结束标记信息给前端, 前端会根据这个标记来判断是否继续接收消息,改变按钮状态从停止生成改为待发送状态
     */
    public void sendEndMarker(WebSocketSession session) {
        try {
            session.sendMessage(new TextMessage("[END_OF_MESSAGE_GENERATE]"));
        } catch (IOException e) {
            log.error("发送结束标记失败", e);
        }
    }

    /**
     * 处理异常
     */
    public void handleError(WebSocketSession session, Exception e) {
        log.error("请求处理异常", e);
        try {
            session.sendMessage(new TextMessage("系统错误: " + e.getMessage()));
            sendEndMarker(session);
        } catch (IOException ex) {
            log.error("发送错误信息失败", ex);
        }
    }
}
